package com.atguigu.edu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.app.func.DimAsyncFunction;
import com.atguigu.edu.realtime.app.func.DorisSinkFunction;
import com.atguigu.edu.realtime.beans.DwsTradeCourseOrderWindowBean;
import com.atguigu.edu.realtime.beans.DwsTradeSourceOrderWindowBean;
import com.atguigu.edu.realtime.utils.DateFormatUtil;
import com.atguigu.edu.realtime.utils.MyKafkaUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * 交易域课程粒度下单统计表
 */
public class DwsTradeCourseOrderWindow {
    public static void main(String[] args) {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(5);
        //TODO 2.检查点相关的设置
        env.enableCheckpointing(5000L);
        //TODO 3.从kakfa主题中读取数据
        //3.1 声明消费的主题以及消费者组
        String topic = "dwd_trade_order_detail";
        String groupId = "dws_trade_course_order_window";
        //3.2 创建消费者对象
        KafkaSource<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, groupId);
        //3.3 消费数据 封装为流
        DataStreamSource<String> kafkaStrDS
                = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka_source");

        //TODO 4.对流中数据进行类型转换并过滤掉空消息  jsonStr->jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String jsonStr, Context ctx, Collector<JSONObject> out) throws Exception {
                        if (StringUtils.isNotEmpty(jsonStr)) {
                            JSONObject jsonObj = JSON.parseObject(jsonStr);
                                out.collect(jsonObj);
                        }
                    }
                }
        );

        //jsonObjDS.print(">>>");
        //TODO 5.去重
        //5.1 按照唯一键(订单明细id)进行分组
        KeyedStream<JSONObject, String> orderDetailIdKeyedDS = jsonObjDS.keyBy(jsonObj -> jsonObj.getString("id"));
        //5.3 去重方式 : 状态 + 抵消
        SingleOutputStreamOperator<JSONObject> distinctDS = orderDetailIdKeyedDS.process(

                new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                    private ValueState<JSONObject> lastJsonObjState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<JSONObject> valueStateDescriptor
                                = new ValueStateDescriptor<JSONObject>("lastJsonObjState", JSONObject.class);
                        valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.seconds(10)).build());
                        lastJsonObjState = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, Context ctx, Collector<JSONObject> out) throws Exception {
                        //判断是否重复
                        JSONObject lastJsonObj = lastJsonObjState.value();

                        if (lastJsonObj != null) {
                            //说明重复了，将状态中影响到度量值的字段进行取反，再发送到下游进行抵消
                            String OriginalAmount = lastJsonObj.getString("origin_amount");
                            String CouponAmount = lastJsonObj.getString("coupon_reduce");
                            String orderTotalAmount = lastJsonObj.getString("final_amount");


                            //取反，发送到下游去
                            lastJsonObj.put("origin_amount", "-" + OriginalAmount);
                            lastJsonObj.put("coupon_reduce", "-" + CouponAmount);
                            lastJsonObj.put("final_amount", "-" +orderTotalAmount);
                            out.collect(lastJsonObj);

                        }
                        lastJsonObjState.update(jsonObj);
                        out.collect(jsonObj);
                    }
                }
        );
        //distinctDS.print("%%%%");
        SingleOutputStreamOperator<DwsTradeCourseOrderWindowBean> beanDS = distinctDS.map(new MapFunction<JSONObject, DwsTradeCourseOrderWindowBean>() {
            @Override
            public DwsTradeCourseOrderWindowBean map(JSONObject jsonObj) throws Exception {

                return DwsTradeCourseOrderWindowBean.builder()
                        .courseId(jsonObj.getString("course_id"))
                        .orderTotalAmount(new BigDecimal(jsonObj.getString("final_amount")))
                        .ts(jsonObj.getLong("ts") * 1000)
                        .build();
            }
        });

        //TODO 7.设置水位线
        SingleOutputStreamOperator<DwsTradeCourseOrderWindowBean> withWaterMarkStream
                = beanDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<DwsTradeCourseOrderWindowBean>forBoundedOutOfOrderness(Duration.ofSeconds(5L))
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<DwsTradeCourseOrderWindowBean>() {
                                    @Override
                                    public long extractTimestamp(DwsTradeCourseOrderWindowBean element, long recordTimestamp) {
                                        return element.getTs();
                                    }
                                }));
        //TODO 8.按照 courseId 分组
        KeyedStream<DwsTradeCourseOrderWindowBean, String> courseIdStream = withWaterMarkStream.keyBy(
                new KeySelector<DwsTradeCourseOrderWindowBean, String>() {
                    @Override
                    public String getKey(DwsTradeCourseOrderWindowBean value) throws Exception {

                        return value.getCourseId();
                    }
                });
        //courseIdStream.print(">>>");
        //TODO 9.开窗聚合
        SingleOutputStreamOperator<DwsTradeCourseOrderWindowBean> reduceStream = courseIdStream.window(
                        TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(2L)))
                .reduce(new ReduceFunction<DwsTradeCourseOrderWindowBean>() {
                    @Override
                    public DwsTradeCourseOrderWindowBean reduce(DwsTradeCourseOrderWindowBean value1, DwsTradeCourseOrderWindowBean value2) throws Exception {
                        value1.setOrderTotalAmount(value1.getOrderTotalAmount().add(value2.getOrderTotalAmount()));
                        //System.out.println(value1);
                        return value1;
                    }
                }, new ProcessWindowFunction<DwsTradeCourseOrderWindowBean, DwsTradeCourseOrderWindowBean, String, TimeWindow>() {
                    @Override
                    public void process(String s, Context context, Iterable<DwsTradeCourseOrderWindowBean> elements, Collector<DwsTradeCourseOrderWindowBean> out) throws Exception {

                        String stt = DateFormatUtil.toYmdHms(context.window().getStart());
                        String edt = DateFormatUtil.toYmdHms(context.window().getEnd());
                        String curDate = DateFormatUtil.toDate(context.window().getStart());
                        for (DwsTradeCourseOrderWindowBean element : elements) {
                            element.setStt(stt);
                            element.setEdt(edt);
                            element.setTs(System.currentTimeMillis());
                            element.setCurDate(curDate);
                            out.collect(element);
                        }
                    }
                });
        //reduceStream.print("$$$");
        //TODO 10.维度关联补充维度数据
        // 10.1 关联课程名称和科目ID
        SingleOutputStreamOperator<DwsTradeCourseOrderWindowBean> courseNameBeanStream = AsyncDataStream.unorderedWait(
                reduceStream,
                new DimAsyncFunction<DwsTradeCourseOrderWindowBean>("dim_course_info") {

                    @Override
                    public void join(JSONObject jsonObject, DwsTradeCourseOrderWindowBean bean) {
                        bean.setCourseName(jsonObject.getString("course_name"));
                        bean.setSubjectId(jsonObject.getString("subject_id"));
                    }

                    @Override
                    public String getKey(DwsTradeCourseOrderWindowBean obj) {
                        return obj.getCourseId();
                    }
                },
                60, TimeUnit.SECONDS);
       // courseNameBeanStream.print("&&&");

        //关联科目表
        SingleOutputStreamOperator<DwsTradeCourseOrderWindowBean> subjectNameBeanStream
                = AsyncDataStream.unorderedWait(courseNameBeanStream, new DimAsyncFunction<DwsTradeCourseOrderWindowBean>("dim_base_subject_info") {

            @Override
            public void join(JSONObject jsonObject, DwsTradeCourseOrderWindowBean bean) {
                bean.setSubjectName(jsonObject.getString("subject_name"));
                bean.setCategoryId(jsonObject.getString("category_id"));
            }

            @Override
            public String getKey(DwsTradeCourseOrderWindowBean obj) {
                return obj.getSubjectId();
            }
        }, 60, TimeUnit.SECONDS);

        // 关联类别表
        SingleOutputStreamOperator<DwsTradeCourseOrderWindowBean> categoryNameBeanStream
                = AsyncDataStream.unorderedWait(subjectNameBeanStream, new DimAsyncFunction<DwsTradeCourseOrderWindowBean>("dim_base_category_info") {

            @Override
            public void join(JSONObject jsonObject, DwsTradeCourseOrderWindowBean bean) {
                bean.setCategoryName(jsonObject.getString("category_name"));
            }

            @Override
            public String getKey(DwsTradeCourseOrderWindowBean obj) {
                return obj.getCategoryId();
            }
        }, 60 , TimeUnit.SECONDS);

        categoryNameBeanStream.print("***");

        // TODO 11 写出到Doris
        categoryNameBeanStream.addSink(new DorisSinkFunction<DwsTradeCourseOrderWindowBean>("dws_trade_course_order_window"));

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
